package com.youyu.core.jms.kafka;

import com.youyu.core.jms.kafka.model.ConsumerData;
import org.yx.exception.BizException;
import org.yx.log.Log;
import org.yx.util.CollectionUtils;

/**
 * <p>Author:  dxj</p>
 * <p>Date:2017/8/28 14:59</p>
 * <p>Description:暂时废弃</p>
 * <p>Copyright: Copyright (c) 2017</p>
 * <p>Company: youyu</p>
 */
@Deprecated
public abstract class AbstractConsumer extends Thread implements IConsumer{

    /**
     * 设置主题
     * @return
     * @throws BizException
     */
    public abstract String setTopic() throws BizException;

    /**
     * 处理核心业务逻辑
     * @param data
     * @throws BizException
     */
    public abstract void process(ConsumerData data) throws BizException;

    @Override
    public void run(){
        long curOffset = -1;
        while(true){
            Customer c = null;
            try{
                String topic = setTopic();
                c = new Customer(topic,  0, curOffset);
                c.open();
                ConsumerData data = c.fetch();
                curOffset = data.getOffset();
                if(CollectionUtils.isNotEmpty(data.getDatas())) {
                    process(data);
                }
            }catch(Exception e){
                Log.get(AbstractConsumer.class).error("消费消息失败："+e.getMessage());
            }finally{
                if(c != null){
                    c.close();
                }
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Log.get(AbstractConsumer.class).error("消费消息中断异常："+e.getMessage());
            }
        }
    }
}
